package com.example.controller;

import com.alibaba.fastjson.JSONObject;
import com.example.dto.Response;
import com.example.enums.EventType;
import com.example.kafka.KafkaConsumerService;
import com.example.kafka.KafkaProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Objects;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    private final KafkaProducerService kafkaProducerService;
    private final KafkaConsumerService kafkaConsumerService;

    @Autowired
    public KafkaController(KafkaProducerService kafkaProducerService, KafkaConsumerService kafkaConsumerService) {
        this.kafkaProducerService = kafkaProducerService;
        this.kafkaConsumerService = kafkaConsumerService;
    }

    @PostMapping("/send/producer")
    public Response<String> sendMessageByProducer(@RequestBody JSONObject message) {
        Integer eventType = message.getInteger("eventType");
        kafkaProducerService.sendMessageToTopic(Objects.requireNonNull(EventType.of(eventType)), message.toJSONString());
        return Response.success("sendMessageByProducer success");
    }

    @PostMapping("/send/consumer")
    public Response<String> sendMessageToConsumer(@RequestBody JSONObject message) {
        kafkaConsumerService.handleSingleMessage(message.toJSONString());
        return Response.success("sendMessageToConsumer success");
    }

}
